package org.my.hadoop.test;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;



public class Deldup {
    public static class TokenizerMapper extends Mapper<Object,Text,Text,Text>{
    	private Text datetext = new Text();
    	public void map(Object key,Text value,Context context) throws IOException, InterruptedException{
    		StringTokenizer line = new StringTokenizer(value.toString(),"\n");
    		while(line.hasMoreTokens()){
    			StringTokenizer itr = new StringTokenizer(line.nextToken());
    			datetext.set(itr.nextToken());
    			Text data = new Text();
    			data.set(itr.nextToken());
    			context.write(datetext, data);
    		}
    	}
    }
    
    public static class DeldupReducer extends Reducer<Text, Text, Text, Text>{
    	public void reduce(Text key,Iterable<Text> values,Context context) throws IOException, InterruptedException{
    		Map<String, Integer> hasMap = new HashMap<String, Integer>();
    		for(Text val:values){
    			if(hasMap.get(val.toString())==null){
    				context.write(key, val);
    				hasMap.put(val.toString(), 1);
    			}
    		}
    	}
    	
    }
    
    
    public static void main(String[] args) throws Exception {
		Configuration configuration = new Configuration();
        if(args.length<2){
        	System.out.println("Usage:Deldup <in>[<in>...] <out>");
        	System.exit(2);
        }
        Job job = Job.getInstance(configuration,"Del dup");
        job.setJarByClass(Deldup.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(DeldupReducer.class);
        job.setReducerClass(DeldupReducer.class);
        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        
        for(int i=0;i<args.length-1;i++){
        	FileInputFormat.addInputPath(job, new Path(args[i]));
        }
        FileOutputFormat.setOutputPath(job, new Path(args[args.length-1]));
        System.exit(job.waitForCompletion(true)?0:1);
	} 
}
